package com.song.hbasechap_7_6;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.sun.jersey.core.impl.provider.entity.XMLJAXBElementProvider.Text;

public class IndexMain extends Configured implements Tool {

	public static final String TEMP_INDEX_PATH = "hdfs://macsong:9000/out/chap_7_6";

	public static void main(String[] args) {
		
		
		
		try {
			ToolRunner.run( new IndexMain(), args);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

	@Override
	public int run(String[] arg0) throws Exception {
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.master", "macsong:60000");
		conf.set("hbase.zookeeper.quorum", "macsong");
		conf.setBoolean("mapred.map.tasks.speculative.execution", false);
		conf.setBoolean("mapred.reduce.tasks.speculative.execution", false);
		Path tempIndexPath = new Path(TEMP_INDEX_PATH);

		Scan scan = new Scan();
		scan.setMaxVersions();

		try {
			FileSystem fs = FileSystem.get(conf);
			if (fs.exists(tempIndexPath)) {
				fs.delete(tempIndexPath, true);
			}

		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		Job job = new Job(conf, "chap_7_6 index JOb");
		TableMapReduceUtil.initTableMapperJob("data_char_7_6", scan,
				IndexMapper.class, Text.class, Text.class, job);
		job.setNumReduceTasks(0);
		job.setOutputFormatClass(TextOutputFormat.class);
		FileOutputFormat.setOutputPath(job, tempIndexPath);
		int success = job.waitForCompletion(true) ? 0 : 1;
		if(success == 0)
		{
			job = new Job(conf, "chap_7_6  load data job");
			
			job.setJarByClass(IndexMain.class);
			
			job.setMapperClass(LoadDataMapper.class);
			job.setNumReduceTasks(0);
			job.setOutputFormatClass(TableOutputFormat.class);
			job.setOutputKeyClass(ImmutableBytesWritable.class);
			job.setOutputValueClass(Writable.class);
			job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, "data_char_7_6_index");
			FileInputFormat.addInputPath(job, tempIndexPath);
			success = job.waitForCompletion(true) ? 0 :1;
		}

		return success;
	}
}
